package com.intramirror.product.schedule.consumer;

import com.intramirror.product.schedule.KafkaItem;
import com.intramirror.product.service.FacilityService;
import java.util.Collections;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Created on 2018/4/10.
 * @author 123
 */
public class ConsumerThread implements Runnable {
    private static final Logger LOGGER = LoggerFactory.getLogger(ConsumerThread.class);

    private FacilityService facilityService;
    private KafkaItem item;

    public ConsumerThread(KafkaItem item, FacilityService facilityService) {
        this.item = item;
        this.facilityService = facilityService;
    }

    @Override
    public void run() {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, item.getInterval());
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, item.getAuto_commit());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, item.getGroup());
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, item.getServers());
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, item.getKey_deserializer());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, item.getValue_deserializer());

        LOGGER.info("Start to receive kafka message with {}.", properties);
        KafkaConsumer consumer = new KafkaConsumer(properties);

        consumer.subscribe(Collections.singletonList(item.getTopic()));
        while (true) {
            try {
                ConsumerRecords<String, String> records = consumer.poll(1024);
                for (ConsumerRecord<String, String> record : records) {
                    LOGGER.info("Kafka received message,partition:{}, offset:{}, topic:{} key:{}, value:{}.", record.partition(), record.offset(),
                            record.topic(), record.key(), record.value());

                    if (item.getConsumer() == 1 || item.getConsumer() == 2 || item.getConsumer() == 3 || item.getConsumer() == 4) {
                        ConsumerFactory.getConsumer(item.getConsumer(), facilityService).process(record.value());
                    } else {
                        LOGGER.warn("Unsupported consumer id {}, topic {}, value {}.", item.getConsumer(), record.topic(), record.value());
                    }
                }
                consumer.commitSync();
            } catch (Exception e) {
                LOGGER.warn("Kafka consumer exception", e);
            }
        }
    }

}
